package com.yanggu.flink.datastream_api.state.keyedstate

import com.yanggu.flink.datastream_api.pojo.SensorReading
import org.apache.flink.api.common.functions.RichFlatMapFunction
import org.apache.flink.api.common.state.ValueStateDescriptor
import org.apache.flink.configuration.Configuration
import org.apache.flink.streaming.api.scala._
import org.apache.flink.util.Collector

/**
 *  flink中的KeyedState, KeyedState只能在KeyedStream中使用, 也就是在keyBy之后
 *  当前数据只能访问自己的key对应的state
 *  常用的状态有ValueState[T]、ListState[T]、MapState[K, V]、ReducingState[T]、AggregatingState[I, O]
 */
object KeyedStateDemo {

  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    env.setParallelism(1)

    env.socketTextStream("localhost", 9000)
      .map(data => {
        val strings = data.split(",")
        SensorReading(strings(0), strings(1).toLong, strings(2).toDouble)
      })
      .keyBy(_.id)
      //map的有状态算子, 只能是在KeyedStream中使用
      .mapWithState[String, Double] {
        //这里实际上使用的是偏函数的简写形式, 模式匹配
        case (data, None) => ("", Some(data.temperature))
        case (data, Some(lastTemperature)) =>
          val diff = (data.temperature - lastTemperature).abs
          var value = ""
          if (diff > 10.0) {
            value = s"传感器编号: ${data.id}, 上次的温度: $lastTemperature, 本次的温度: ${data.temperature}, 温度的差值: $diff"
          }
          (value, Some(data.temperature))
      }
      .filter(_.nonEmpty)
      //可以自定义类实现对应的RichFunction, 一般而言都是实现RichFlatMapFunction, 比较灵活可以实现map和filter的功能
      //.flatMap(new MyRichFlatMapFunction1(10.0D))
      .print("传感器的异常温度检测")

    env.execute("state test")
  }

}

class MyRichFlatMapFunction1(val diff: Double) extends RichFlatMapFunction[SensorReading, String] {

  /**
   * scala的lazy关键字可以实现懒加载，并且在多线程的情况下保证线程安全,
   * lazy关键字只能修饰val不可变的类型, 同时状态需要和上下文进行绑定
   */
  private lazy val lastTempState = getRuntimeContext.getState(
    new ValueStateDescriptor[Option[Double]]("lastTempState", classOf[Option[Double]]))

  override def open(parameters: Configuration): Unit = {
    //可以在open方法中初始化状态
    //lastTempState = getRuntimeContext.getState(
    //  new ValueStateDescriptor[Option[Double]]("lastTempState", classOf[Option[Double]]))
    //在open方法中不能使用状态。例如使用update()和value()方法都会报错
  }

  override def flatMap(data: SensorReading, out: Collector[String]): Unit = {
    //检测某个传感器前后两个温度的差值是否大于设置值
    val option = lastTempState.value()
    //option为null表示状态中没有数据
    if (option != null) {
      val lastTemperature = option.get
      val abs = (data.temperature - lastTemperature).abs
      if (abs > diff) {
        out.collect(s"传感器编号: ${data.id}, 上次温度: ${lastTemperature}, 本次温度: ${data.temperature}, 温度差值: $abs")
      }
    }
    lastTempState.update(Some(data.temperature))
  }

}
